package collect

import (
	"context"
	"runtime"
	"sync"
	"sync/atomic"
	"time"
	"tinyGW/app/api/repository"
	"tinyGW/app/models"
	"tinyGW/pkg/service/collect/channel"
	"tinyGW/pkg/service/collect/worker"
	"tinyGW/pkg/service/command"
	"tinyGW/pkg/service/conf"
	"tinyGW/pkg/service/event"

	"go.uber.org/zap"
)

type (
	// CollectorServer 采集接口服务器
	CollectorServer struct {
		collectors           *sync.Map
		deviceRepository     repository.DeviceRepository
		deviceTypeRepository repository.DeviceTypeRepository
		config               *conf.Config
		eventSrv             *event.EventService

		// 并发控制和资源管理
		workerPool    *WorkerPool
		healthChecker *HealthChecker

		// 性能指标
		metrics CollectorMetrics

		// 控制信号
		ctx    context.Context
		cancel context.CancelFunc

		commandManager *command.Manager
	}

	// WorkerPool 工作器池管理
	WorkerPool struct {
		maxWorkers    int
		activeWorkers int32
		workerLimiter chan struct{}
		workerQueue   chan func()
	}

	// HealthChecker 健康检查器
	HealthChecker struct {
		checkInterval time.Duration
		unhealthyMap  *sync.Map // 存储不健康的采集器
		lastCheckTime int64
	}

	// CollectorMetrics 采集器性能指标
	CollectorMetrics struct {
		TotalCollectors  int32 `json:"total_collectors"`
		ActiveCollectors int32 `json:"active_collectors"`
		FailedCollectors int32 `json:"failed_collectors"`
		TotalWorkers     int32 `json:"total_workers"`
		ActiveWorkers    int32 `json:"active_workers"`
		EventsProcessed  int64 `json:"events_processed"`
		HealthCheckCount int64 `json:"health_check_count"`
	}
)

// NewCollectorServer 实例化
func NewCollectorServer(
	deviceRepository repository.DeviceRepository,
	config *conf.Config,
	e *event.EventService,
	deviceTypeRepository repository.DeviceTypeRepository,
	commandManager *command.Manager,
) *CollectorServer {
	// 根据CPU核心数和配置决定最大工作线程数
	maxWorkers := runtime.NumCPU() * 2
	if config.Server.MaxWorkers > 0 {
		maxWorkers = config.Server.MaxWorkers
	}

	// 确保最少有4个工作线程，最多不超过64个
	if maxWorkers < 4 {
		maxWorkers = 4
	} else if maxWorkers > 64 {
		maxWorkers = 64
	}

	zap.S().Infof("初始化采集服务器，最大工作线程: %d, CPU核心数: %d", maxWorkers, runtime.NumCPU())

	ctx, cancel := context.WithCancel(context.Background())

	server := &CollectorServer{
		collectors:           &sync.Map{},
		deviceRepository:     deviceRepository,
		deviceTypeRepository: deviceTypeRepository,
		config:               config,
		eventSrv:             e,
		ctx:                  ctx,
		cancel:               cancel,
		commandManager:       commandManager,

		workerPool: &WorkerPool{
			maxWorkers:    maxWorkers,
			workerLimiter: make(chan struct{}, maxWorkers),
			workerQueue:   make(chan func(), maxWorkers*2), // 队列容量是工作线程的2倍
		},

		healthChecker: &HealthChecker{
			checkInterval: 30 * time.Second, // 30秒检查一次
			unhealthyMap:  &sync.Map{},
		},
	}

	// 启动工作器池
	server.startWorkerPool()

	// 启动健康检查
	server.startHealthChecker()

	return server
}

// startWorkerPool 启动工作器池
func (cs *CollectorServer) startWorkerPool() {
	for i := 0; i < cs.workerPool.maxWorkers; i++ {
		go cs.workerPoolWorker()
	}
}

// workerPoolWorker 工作器池的工作者
func (cs *CollectorServer) workerPoolWorker() {
	atomic.AddInt32(&cs.workerPool.activeWorkers, 1)
	defer atomic.AddInt32(&cs.workerPool.activeWorkers, -1)

	for {
		select {
		case <-cs.ctx.Done():
			return
		case task := <-cs.workerPool.workerQueue:
			if task != nil {
				task()
			}
		}
	}
}

// startHealthChecker 启动健康检查
func (cs *CollectorServer) startHealthChecker() {
	go func() {
		ticker := time.NewTicker(cs.healthChecker.checkInterval)
		defer ticker.Stop()

		for {
			select {
			case <-cs.ctx.Done():
				return
			case <-ticker.C:
				cs.performHealthCheck()
			}
		}
	}()
}

// performHealthCheck 执行健康检查
func (cs *CollectorServer) performHealthCheck() {
	atomic.StoreInt64(&cs.healthChecker.lastCheckTime, time.Now().Unix())
	atomic.AddInt64(&cs.metrics.HealthCheckCount, 1)

	var totalCount, activeCount, failedCount int32

	cs.collectors.Range(func(key, value interface{}) bool {
		totalCount++

		if w, ok := value.(worker.Worker); ok {
			// 检查采集器是否健康（这里可以添加更复杂的健康检查逻辑）
			if cs.isWorkerHealthy(w, key.(string)) {
				activeCount++
				// 从不健康列表中移除
				cs.healthChecker.unhealthyMap.Delete(key)
			} else {
				failedCount++
				// 添加到不健康列表
				cs.healthChecker.unhealthyMap.Store(key, time.Now())
			}
		}

		return true
	})

	// 更新指标
	atomic.StoreInt32(&cs.metrics.TotalCollectors, totalCount)
	atomic.StoreInt32(&cs.metrics.ActiveCollectors, activeCount)
	atomic.StoreInt32(&cs.metrics.FailedCollectors, failedCount)
	atomic.StoreInt32(&cs.metrics.ActiveWorkers, atomic.LoadInt32(&cs.workerPool.activeWorkers))

	if failedCount > 0 {
		zap.S().Warnf("健康检查完成: 总计%d个采集器, 活跃%d个, 失败%d个", totalCount, activeCount, failedCount)
	}
}

// isWorkerHealthy 检查工作器是否健康
func (cs *CollectorServer) isWorkerHealthy(w worker.Worker, name string) bool {
	// 检查任务队列是否过满
	if w.CollectTaskIsFull() {
		zap.S().Warnf("采集器 %s 任务队列已满", name)
		return false
	}

	// 这里可以添加更多健康检查逻辑
	// 比如检查最后执行时间、错误计数等

	return true
}

// InitCollectorServer 初始化
func InitCollectorServer(server *CollectorServer, repository repository.CollectorRepository, e *event.EventService) {
	zap.S().Info("初始化采集接口服务器")

	// 批量加载所有采集器
	if collectors, err := repository.FindAll(); err == nil {
		activeCount := len(collectors)
		zap.S().Infof("总采集器数量: %d, 活跃采集器: %d", len(collectors), activeCount)

		// 使用工作器池并发加载采集器
		for _, collector := range collectors {
			// 使用闭包捕获变量
			c := collector
			server.submitTask(func() {
				server.Add(c)
			})
		}
	}

	// 注册事件处理 - 使用工作器池异步处理
	e.Subscribe("collector_add", func(e event.Event) {
		server.submitTask(func() {
			atomic.AddInt64(&server.metrics.EventsProcessed, 1)
			server.Add(e.Data.(models.Collector))
		})
	})

	e.Subscribe("collector_update", func(e event.Event) {
		server.submitTask(func() {
			atomic.AddInt64(&server.metrics.EventsProcessed, 1)
			server.Update(e.Data.(models.Collector))
		})
	})

	e.Subscribe("collector_delete", func(e event.Event) {
		server.submitTask(func() {
			atomic.AddInt64(&server.metrics.EventsProcessed, 1)
			server.Delete(e.Data.(string))
		})
	})
}

// submitTask 提交任务到工作器池
func (cs *CollectorServer) submitTask(task func()) {
	select {
	case cs.workerPool.workerQueue <- task:
		// 任务成功提交
	default:
		// 队列已满，直接执行（降级处理）
		zap.S().Warn("工作器队列已满，直接执行任务")
		go task()
	}
}

// Add 新增采集接口服务器
func (cs *CollectorServer) Add(collector models.Collector) {
	if len(collector.Name) == 0 {
		zap.S().Error("新增采集接口服务器失败，名称不能为空")
		return
	}

	zap.S().Infof("新增采集接口服务器: %s (类型: %s)", collector.Name, collector.Type)

	// 获取工作器限制器
	select {
	case cs.workerPool.workerLimiter <- struct{}{}:
		defer func() { <-cs.workerPool.workerLimiter }()
	case <-cs.ctx.Done():
		return
	}

	w := worker.NewWorker(
		collector, cs.deviceRepository,
		cs.config, cs.eventSrv,
		cs.deviceTypeRepository,
		cs.commandManager,
	)

	w.Start()
	cs.collectors.Store(collector.Name, w)
	atomic.AddInt32(&cs.metrics.TotalCollectors, 1)
}

// Delete 删除采集接口服务器
func (cs *CollectorServer) Delete(name string) {
	zap.S().Info("删除采集接口服务器", name)
	if value, loaded := cs.collectors.LoadAndDelete(name); loaded {
		if c, ok := value.(worker.Worker); ok {
			c.Stop()
		}
		atomic.AddInt32(&cs.metrics.TotalCollectors, -1)
		// 从不健康列表中移除
		cs.healthChecker.unhealthyMap.Delete(name)
	}
}

// Update 修改采集接口服务器
func (cs *CollectorServer) Update(collector models.Collector) {
	cs.Delete(collector.Name)
	cs.Add(collector)
}

// FindByCollectorName 根据采集器名称查找工作器
func (cs *CollectorServer) FindByCollectorName(collectorName string) (worker.Worker, bool) {
	if value, ok := cs.collectors.Load(collectorName); ok {
		if work, ok := value.(worker.Worker); ok {
			return work, true
		}
	}
	return nil, false
}

// FindByDeviceName 根据设备名称查找工作器
func (cs *CollectorServer) FindByDeviceName(deviceName string) (worker.Worker, bool) {
	device, err := cs.deviceRepository.Find(deviceName)
	if err != nil {
		return nil, false
	}
	if value, ok := cs.collectors.Load(device.Collector.Name); ok {
		if work, ok := value.(worker.Worker); ok {
			return work, true
		}
	}
	return nil, false
}

// GetMetrics 获取服务器指标
func (cs *CollectorServer) GetMetrics() CollectorMetrics {
	metrics := cs.metrics
	metrics.ActiveWorkers = atomic.LoadInt32(&cs.workerPool.activeWorkers)
	return metrics
}

// GetChannelMetrics 获取所有采集器的通道指标
func (cs *CollectorServer) GetChannelMetrics() map[string]channel.ChannelMetrics {
	result := make(map[string]channel.ChannelMetrics)

	cs.collectors.Range(func(key, value interface{}) bool {
		if _, ok := key.(string); ok {
			if _, ok := value.(worker.Worker); ok {
				// 这里需要Worker接口提供GetChannelMetrics方法
				// result[name] = w.GetChannelMetrics()
				// 暂时返回空结果，待Worker接口扩展后实现
			}
		}
		return true
	})

	return result
}

// GetUnhealthyCollectors 获取不健康的采集器列表
func (cs *CollectorServer) GetUnhealthyCollectors() map[string]time.Time {
	result := make(map[string]time.Time)

	cs.healthChecker.unhealthyMap.Range(func(key, value interface{}) bool {
		if name, ok := key.(string); ok {
			if timestamp, ok := value.(time.Time); ok {
				result[name] = timestamp
			}
		}
		return true
	})

	return result
}

// Shutdown 优雅关闭服务器
func (cs *CollectorServer) Shutdown() {
	zap.S().Info("开始关闭采集服务器...")

	// 取消上下文，停止所有后台任务
	cs.cancel()

	// 停止所有采集器
	cs.collectors.Range(func(key, value interface{}) bool {
		if name, ok := key.(string); ok {
			if w, ok := value.(worker.Worker); ok {
				zap.S().Infof("停止采集器: %s", name)
				w.Stop()
			}
		}
		return true
	})

	zap.S().Info("采集服务器已关闭")
}
